// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu, jhkim

#include "coin2/strategy/linear_model/product_strategy.h"

#include <experimental/filesystem>
#include <utility>
#include <vector>
#include <unordered_set>

#include <boost/exception/diagnostic_information.hpp>

#include "coin2/base/feed_convert_fastfeature.h"
#include "coin2/base/log.h"
#include "coin2/base/time.h"
#include "coin2/exchange/base/api_util/raw_subscriber_util.h"
#include "coin2/exchange/factory/symbology/product.h"
#include "coin2/exchange/symbology.pb.h"
#include "coin2/strategy/linear_model/logic_factory.h"
#include "coin2/strategy/util/alarm.h"

namespace coin2::strategy::linear_model {

using ::coin2::base::ConvertFeedMessage;
using ::coin2::base::LogEveryDuration;

const Symbol* GetSymbolInline(
    const SymbolManager* symbol_manager,
    const std::string& ex,
    const std::string& norm) {
  return symbol_manager->FromString(ex, norm);
}

void FeatureFeedConverter::RegisterSymbols() {
  std::vector<const Symbol*> outs;
  symbol_manager->GetAllSymbols(&outs);
  for (auto& out : outs) {
    symbol_halt_.AddSymbol(out);
    CHECK(out);
    std::string exchange_coin2 = replace_string(out->exchange_str(), "Futures", "");
    ExchangeType exchange;
    CHECK(coin::proto::ExchangeType_Parse(exchange_coin2, &exchange));

    MarketType market = MarketType::Spot;

    if (exchange == coin::proto::ExchangeType::Bitmex) {
      market = MarketType::Futures;
    } else if (exchange_coin2 != out->exchange_str()) {
      market = MarketType::Futures;
    }
    auto em_key = std::make_tuple(exchange, market);
    if (em_map_.count(em_key) > 0) {
      CHECK_EQ(em_map_.at(em_key), out->exchange_str());
    } else {
      em_map_[em_key] = out->exchange_str();
      LOG(INFO) << "Registered " << MarketType_Name(market) << "."
                << coin::proto::ExchangeType_Name(exchange) << " : " << out->exchange_str();
    }
  }
}


FeatureFeedConverter::FeatureFeedConverter(
    const SymbolManager* symbol_manager,
    ::coin2::exchange::base::symbology::ProductEncyclopedia* product_cache,
    bool register_symbols,
    std::optional<int64_t> tfeed_etft_guard_ns)
    : symbol_manager(symbol_manager),
      tfeed_etft_guard_ns(tfeed_etft_guard_ns),
      product_cache_(product_cache) {
  if (register_symbols) {
    std::vector<const Symbol*> outs;
    symbol_manager->GetAllSymbols(&outs);
    RegisterSymbols();
  }
}

const Symbol* FeatureFeedConverter::GetSymbol(const FeedUpdate& upd) {
  const Symbol* symbol = nullptr;
  if (upd.mea().exchange == coin::proto::ExchangeType::Bitflyer &&
      StringEndsWith(upd.product().relative_norm(), "IMMEDIATE")) {
    symbol = GetSymbolInline(symbol_manager, "Bitflyer", upd.product().relative_norm());
  } else {
    if (!HasMea(upd.mea())) {
      return nullptr;
    }
    const std::string& ex = GetCoin1ExchangeName(upd.mea());
    symbol = GetSymbolInline(symbol_manager, ex, upd.product().relative_norm());
  }
  if (symbol_manager->IsBanSymbol(symbol)) return nullptr;
  return symbol;
}

void FeatureFeedConverter::onTradeFeed(const FeedUpdate& upd) {
  feed_converted = false;
  // FT - ET > 2: reject to convert T feed
  if (tfeed_etft_guard_ns) {
    if (upd.trade().timestamp + *tfeed_etft_guard_ns < upd.timestamp()) {
      return;
    }
  }
  onFeed(GetSymbol(upd), upd);
}

void FeatureFeedConverter::onBookFeed(const FeedUpdate& upd) {
  feed_converted = false;
  if (upd.exchange() == ExchangeType::Bithumb
      && upd.book().Ask0()->price > upd.book().Bid0()->price * 1.1) {
    return;
  }
  const Symbol* symbol = GetSymbol(upd);
  if (upd.book().Ask0() && upd.book().Bid0()) {
    onFeed(symbol, upd);
  }
}

void FeatureFeedConverter::onStatusFeed(const FeedUpdate& upd) {
  feed_converted = false;
  onFeed(GetSymbol(upd), upd);
}

void FeatureFeedConverter::onFeed(const Symbol* symbol, const FeedUpdate& upd) {
  before_feature_calc_time = GetCurrentTimestamp();
  feed_converted = false;
  if (symbol == nullptr) {
    return;
  }
  last_feed_msg.Clear();
  ConvertFeedMessage(upd, 5, &last_feed_msg, qty_multiplier);

  double price_ticksize = 0;
  try {
    auto&& pholder = product_cache_->holder(upd.product());
    switch (last_feed_msg.feed_type()) {
      case ::fastfeature::BOOK_FEED:
        price_ticksize = pholder.product_info().tick_price()
            .UpTicksize(upd.book().Ask0()->price);
        last_feed_msg.set_ticksize(price_ticksize);
        break;
      case ::fastfeature::TRADE_FEED:
        price_ticksize = pholder.product_info().tick_price()
            .UpTicksize(upd.trade().price);
        last_feed_msg.set_ticksize(price_ticksize);
            default:
        break;
    }
  } catch (const std::runtime_error& error) {
    LOG(ERROR) << error.what();
    return;
  }
  last_feed_msg.set_is_halt(
      symbol_halt_.IsHalt(symbol, upd.timestamp()));
  feed_converted = true;
}

void ProductLmStrategy::onTradeFeed(const FeedUpdate& upd) {
  // TODO(jhkim): slimfeed bugreport
  if (upd.trade().price <= 0) {
    LOG(ERROR) << upd.trade();
    return;
  }
  converter->onTradeFeed(upd);

  if (converter->feed_converted) {
    onFeatureFeed(upd, &converter->last_feed_msg, simulation_);
    linear_model_logic_->WriteTelemetry(upd);
  }
}

void ProductLmStrategy::onBookFeed(const FeedUpdate& upd) {
  converter->onBookFeed(upd);

  if (converter->feed_converted) {
    onFeatureFeed(upd, &converter->last_feed_msg, simulation_);
    linear_model_logic_->WriteTelemetry(upd);
  }
}

void ProductLmStrategy::onStatusFeed(const FeedUpdate& upd) {
  converter->onStatusFeed(upd);
  if (converter->feed_converted) {
    onFeatureFeed(upd, &converter->last_feed_msg, simulation_);
    linear_model_logic_->WriteTelemetry(upd);
  }
}

void ProductLmStrategy::onFeatureFeed(
    const FeedUpdate& upd,
    const ::fastfeature::FeedMessage* feed_msg,
    bool simulation) {
  const Symbol* symbol = converter->GetSymbol(upd);
  // this linear model substrategy won't take this symbol.
  if (symbol == nullptr) {
    return;
  }
  if (!target_product_) {
    printf("product feed not given yet..\n");
    return;
  }

  std::optional<int64_t> after_feature_calc_time;
  if (!linear_model_logic_->BanUpdatingFeatureFeed(symbol)) {
    linear_model_logic_->UpdateModelFeedOnly(
        symbol,
        *feed_msg,
        lm_feed_id_,
        global_feed_id_,
        linear_model_);
    after_feature_calc_time = GetCurrentTimestamp();
  }

  if (strat_logger_) {
    strat_logger_->before_feature_calc_time = converter->before_feature_calc_time;
    strat_logger_->after_feature_calc_time = after_feature_calc_time;
  }

  if (force_try_clear_) {
    linear_model_logic_->TryClear(symbol, upd, *feed_msg);
    return;
  }

  if (ftet_guard_) {
    switch (upd.mea().market) {
      case MarketType::Futures:
        warmup_.UpdateFeed(symbol, *feed_msg, 120L * 1e9);
        break;
      case MarketType::Spot:
        warmup_.UpdateFeed(symbol, *feed_msg, 500L * 1e9);
        break;
      default:
        break;
    }
    const Symbol* blocked = warmup_.IsBlockedAny(*feed_msg);
    if (blocked) {
      if (LogEveryDuration(&lastlogt_blocked_, feed_msg->fetched_time(), 10'000'000'000L)) {
        LOG(INFO)
            << "[target_strat]: " << target_->symbol_string()
            << ", symbol: " << blocked->symbol_string()
            << " feed BLOCKED not ready.";
      }
      return;
    }

    const Symbol* outdated = warmup_.IsOutdatedAny(feed_msg->fetched_time());
    if (outdated) {
      if (LogEveryDuration(&lastlogt_outdated_, feed_msg->fetched_time(), 10'000'000'000L)) {
        LOG(INFO)
            << "[target_strat]: " << target_->symbol_string()
            << ", symbol: " << outdated->symbol_string()
            << " feed OUTDATED not ready.";
      }
      int mulp = 1;
      if (outdated->symbol_string().find("OkexFutures") != std::string::npos &&
          outdated->symbol_string().find("PERPETUAL") != std::string::npos) {
        mulp = 3;
      }
      if (warmup_.DurationOutdated(feed_msg->fetched_time()) > kDefaultClearTimeout_ * mulp) {
        linear_model_logic_->TryClear(symbol, upd, *feed_msg);
      }
      return;
    }

    if (alarm_ && alarm_->state == util::AlarmState::ALARM_ON) {
      if (LogEveryDuration(&lastlogt_alarm_, feed_msg->fetched_time(), 10'000'000'000L)) {
        LOG(ERROR) << "[target_strat]: " << target_->symbol_string() << ", ALARM on";
      }
      return;
    }

    halt_manager_.UpdateTime(upd.timestamp());
    if (halt_manager_.halt) {
      if (LogEveryDuration(&lastlogt_warning_, feed_msg->fetched_time(), 10'000'000'000L)) {
        LOG(WARNING) << "[target_strat]: " << target_->symbol_string() << ", trading halt";
      }
      return;
    }
  }

  linear_model_logic_->onFeed(
      symbol,
      upd,
      *feed_msg,
      linear_model_,
      lm_feed_id_,
      global_feed_id_,
      delegate_,
      simulation);
}

ProductLmStrategy::ProductLmStrategy(
    LinearModel* linear_model,
    int64_t* lm_feed_id,
    const int64_t* global_feed_id,
    LinearModelDelegateProto* delegate,
    const MarketExchangeApi& mea,
    std::string symbol_str,
    const LmProductConfig& lm_product_config,
    ::coin2::exchange::base::symbology::ProductEncyclopedia* product_cache,
    int64_t timestamp_override,
    bool simulation,
    const VolumeInfo& volume_info,
    const std::vector<presto::Mep>& clear_meps,
    std::shared_ptr<::fastfeature::KlineInfoProto> kline_info = nullptr)
    : linear_model_(linear_model),
      lm_feed_id_(lm_feed_id),
      global_feed_id_(global_feed_id),
      delegate_(delegate),
      simulation_(simulation),
      ftet_guard_(
          lm_product_config.profile_name().find("treemulti") != std::string::npos &&
          !lm_product_config.has_focus_config()),
      force_try_clear_(lm_product_config.force_try_clear()),
      halt_manager_(lm_product_config.halt_config()),
      timestamp_override_(timestamp_override) {
  {
    ::coin2::exchange::symbology::ProductConfig config;
    if (symbol_str.find("*") != std::string::npos) {
      config.set_norm_format(symbol_str);
    } else {
      config.add_norms(symbol_str);
    }
    auto products = ProductRequestsFromConfig(mea.String(), config, timestamp_override);
    target_product_ = std::move(products[0]);
    symbol_str = target_product_->relative_norm();
  }
  LOG(INFO) << lm_product_config.DebugString();

  converter.reset(new FeatureFeedConverter(
      linear_model_->GetFeatureManager()->GetSymbolManager(),
      product_cache,
      true,
      lm_product_config.has_focus_config() ? std::nullopt : std::optional<int64_t>(2e9)));
  std::vector<const Symbol*> targets;
  linear_model_->GetSymbols(&targets);
  target_ = nullptr;
  is_focus_ = lm_product_config.has_focus_config()
    || lm_product_config.has_focus_midfreq_config()
    || lm_product_config.search_focus_midfreq_config_size() > 0;
  if (!lm_product_config.has_interval_config() && !is_focus_) {
    CHECK_EQ(targets.size(), 1);
    target_ = targets.front();
  }
  if (is_focus_) CHECK_EQ(target_, nullptr);

  CHECK_GT(targets.size(), 0) << targets.size();

  const ::presto::Symbol* coin1_target_symbol =
      GetSymbolInline(converter->symbol_manager, converter->GetCoin1ExchangeName(mea), symbol_str);

  linear_model_logic_ = LmLogicFactory::CreateNewLmLogic(
      mea,
      *target_product_,
      coin1_target_symbol,
      lm_product_config,
      product_cache,
      linear_model_->GetFeatureManager()->GetSymbolManager(),
      kline_info);

  std::unordered_set<std::string> clear_product_names;
  for (auto mep : clear_meps) {
    clear_product_names.insert(mep.product_name);
  }
  linear_model_logic_->SetClearProductNames(clear_product_names);
  linear_model_logic_->SetProductStrategy(this);
}

}  // namespace coin2::strategy::linear_model
